package com.zp.self.producer;

import com.zp.self.entity.ProduceMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import javax.validation.constraints.NotBlank;
import java.util.Random;

/**
 * @author By ZengPeng
 * @Description 生产消息的服务实现
 * @date in  2021/3/26 11:21
 * @Modified By
 */
@Slf4j
@Service
public class ProduceServiceImpl implements ProduceService{

    @Resource
    private DefaultMQProducer defaultProducer;

    @Resource
    private TransactionMQProducer transactionProducer;

    /**
    *  异步回调函数
    **/
    private final RocketSendCallback rocketSendCallback = new RocketSendCallback();
    /**
    *  事务回调用对象
    **/
    private final MyTransactionListener mt1 = new MyTransactionListener();

    /**
     * 发送同步消息
     * @param produceMessage 消息详情
     */
    @Override
    public SendResult sendSyncMsg(ProduceMessage produceMessage) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        Message message = createMessage(produceMessage);
        return defaultProducer.send(message);
    }

    /**
     * 发送异步消息
     * @param produceMessage 消息详情
     */
    @Override
    public void sendAsyncMsg(ProduceMessage produceMessage) throws RemotingException, MQClientException, InterruptedException {
        Message message = createMessage(produceMessage);
        defaultProducer.send(message,rocketSendCallback);
    }

    /**
     * 发送事务消息
     * @param produceMessage 消息详情
     */
    @Override
    public SendResult sendTransactionMsg(ProduceMessage produceMessage) throws Exception {
        synchronized (mt1){
            transactionProducer.setTransactionListener(mt1);
            Message message = createMessage(produceMessage);
            transactionProducer.sendMessageInTransaction(message,null);
        }
        return null;
    }

    /**
     * 发送顺序消息
     * @param produceMessage 消息详情
     * @param orderId 以此确认消息队列
     */
    @Override
    public SendResult sendMsgOrder(ProduceMessage produceMessage, int orderId) throws Exception {
        return null;
    }

    /**
     * 单向消息
     * @param produceMessage 消息详情
     */
    @Override
    public void sendOneWayMsg(ProduceMessage produceMessage) throws Exception {

    }



    //---------------------------------- 私有方法 ----------------------------------
    //事务监听对象
    class MyTransactionListener implements  TransactionListener {
        private int flag = 0;

        @Override
        public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
            System.out.println("-----------------现在开始执行业务-----------------");
            flag = new Random().nextInt(3);
            switch (flag) {
                case 0: {
                    System.out.println("当前flag = 0  提交了消息！");
                    return LocalTransactionState.COMMIT_MESSAGE;
                }
                case 1: {
                    System.out.println("当前flag =1   回滚了消息！");
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }
                case 2: {
                    System.out.println("当前flag = 2   不知道啥情况，请会查？！");
                    return LocalTransactionState.UNKNOW;
                }
                default: {
                    System.out.println("系统故障了flag=" + flag);
                    throw new RuntimeException("系统故障了flag=" + flag);
                }
            }
        }

        @Override
        public LocalTransactionState checkLocalTransaction(MessageExt msg) {
            System.out.println("-----------------现在开始回查-----------------");
            flag = new Random().nextInt(2);
            switch (flag) {
                case 0: {
                    System.out.println("当前flag = 0  提交了消息！");
                    return LocalTransactionState.COMMIT_MESSAGE;
                }
                case 1: {
                    System.out.println("当前flag =1   回滚了消息！");
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }
                default: {
                    System.out.println("系统故障了flag=" + flag);
                    throw new RuntimeException("系统故障了flag=" + flag);
                }
            }
        }
    }
    /**
    *  异步回调方法
    **/
    class RocketSendCallback implements SendCallback {
        @Override
        public void onSuccess(SendResult sendResult) {
            log.info("send message success. topic=" + sendResult.getMessageQueue().getTopic() + ", msgId=" + sendResult.getMsgId());
        }
        @Override
        public void onException(Throwable e) {
            log.error("send message failed.", e);
        }
    }


    private Message createMessage(ProduceMessage produceMessage) {
        @NotBlank String topic = produceMessage.getTopic();
        @NotBlank String content = produceMessage.getContent();
        @NotBlank String tag = produceMessage.getTag();
        String keys = produceMessage.getKeys();
        return new Message(topic, tag, keys, content.getBytes());
    }
}
